package com.chukun.flink.dataset.broad;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.configuration.Configuration;

import java.util.List;

/**
 * @author chukun
 * @version 1.0.0
 * @description broadcast基本操作
 * @createTime 2022年05月29日 14:35:00
 */
public class BroadVariableTemplate {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<Integer> broadcast = env.fromElements(1, 2, 3);

        DataSource<String> dataset = env.fromElements("a", "b");

        MapOperator<String, String> mappedDataset = dataset.map(new BroadMapTemplate()).withBroadcastSet(broadcast, "broadcast.dataset");

        mappedDataset.print("withBroadcastSet");

        env.execute("BroadVariableTemplate");
    }

    private static class BroadMapTemplate extends RichMapFunction<String, String> {

        private List<Integer> broadcastCollection;

        @Override
        public String map(String value) throws Exception {
            return value + broadcastCollection.toString();
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            broadcastCollection = getRuntimeContext().getBroadcastVariable("broadcast.dataset");
        }
    }
}
